
#include "config.h"
#include "manager_info.h"
#include <json/json.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <net/if.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <linux/sockios.h>
#include <sys/ioctl.h>
#include <arpa/inet.h>
#include <sys/sysinfo.h>
#include <fstream>
#include "packagefilter.h"
#include "processhub.h"

namespace ProcessRouter {

static void
zk_proxy_pool_data_watcher_g(int rc, const char *value, int value_len,
        const struct Stat *stat, const void *data)
{
    if (rc != 0)
        return;

    std::string content(value, value_len);

    std::list<std::string> proxys;

    while(!content.empty()) {
        size_t pos = content.find("\n");
        if (pos == -1)
            pos = content.size();
        std::string proxy = content.substr(0, pos);
        proxys.push_back(proxy);
        content = content.erase(0, pos+1);
    }

    ProcessHub::instance()->info()->setProxyPool(proxys);
}

static void 
zk_proxy_pool_watcher_g(zhandle_t* zh, int type, int state,
        const char* path, void* watcherCtx)
{
    zoo_aget(zh, path, 1, zk_proxy_pool_data_watcher_g, NULL);
}

ManagerInfo::ManagerInfo()
    : m_enableProxy(false)
    , m_enableProxyPool(false)
    , m_proxyPoolPos(0)
    , m_zkhandle(NULL)
{
    std::string type_name = ProcessRouter::CoreStatus::descriptor()->full_name();
    rooftop_add_event(type_name.c_str()); 
    Rooftop::EventManager::getInstance()->addEventCallback(type_name.c_str(),
            boost::bind(&ManagerInfo::updateCoreStatus, this, _1));

    type_name = ProcessRouter::CorePages::descriptor()->full_name();
    rooftop_add_event(type_name.c_str()); 
    Rooftop::EventManager::getInstance()->addEventCallback(type_name.c_str(),
            boost::bind(&ManagerInfo::updateCorePages, this, _1));
}

ManagerInfo::~ManagerInfo()
{

}

int ManagerInfo::initFromLocalFile(std::string path)
{
    GOOGLE_PROTOBUF_VERIFY_VERSION;

    m_ip = getIPAddress();
    m_infoStruct.set_scurrentpc(0);
    m_infoStruct.set_dcurrentpc(0);
    m_infoStruct.set_soverflowpc(0);
    m_infoStruct.set_doverflowpc(0);
    m_infoStruct.set_coresize(0);
    m_infoStruct.set_vaildcs(0);
    m_infoStruct.set_scoresize(0);
    m_infoStruct.set_dcoresize(0);
    m_infoStruct.set_connectsize(0);

    int fp = open(path.c_str(), O_RDONLY);

    if (fp <= 0) {
        perror("Open configuer file error");
        return -1;
    }

    int ret = -1;
    char buf[1024] = {0};
    std::string content;

    do {
        ret = read(fp, buf, 1023);
        content.append(buf);
    } while (ret >= 1023);

    if (parseConfiguer(content)) {
        close(fp);
        return -1;
    }

    close(fp);

    m_infoStruct.set_protocol("tcp://"+m_ip+":"+configuer()->socketport());

    updateMemInfo();

    return 0;
}

bool ManagerInfo::updateCorePages(Rooftop::EventBase &e)
{
    PackageEvent *event = static_cast<PackageEvent*>(&e);
    if (!event->package().has_pages())
        return false;

    CorePages sta = event->package().pages();

    boost::unique_lock<boost::mutex> lock(m_mutex);
    m_pageInfoMap[sta.pid()] = sta;
}

bool ManagerInfo::updateCoreStatus(Rooftop::EventBase &e)
{
    PackageEvent *event = static_cast<PackageEvent*>(&e);
    if (!event->package().has_status())
        return false;

    CoreStatus sta = event->package().status();

    double level = (sta.runningcount() + sta.waitcount())/mempercent();

    if ((sta.runningcount() + sta.waitcount()) > m_infoStruct.configuer().exitworkcount())
        level += (sta.runningcount() + sta.waitcount()) - m_infoStruct.configuer().exitworkcount();

    sta.set_level(level);

    if (!sta.vaild())
        m_needDestroyProcess.push_back(sta.pid());

    bool haschange = false;
    bool hasnvaild = false;
    for (int i = 0; i < m_infoStruct.status_size(); i++) {
        CoreStatus *status = m_infoStruct.mutable_status(i);
        if (!status->vaild() && status->pid() == sta.pid()) {
            haschange = true;
            boost::unique_lock<boost::mutex> lock(m_mutex);
            *status = sta;
            lock.unlock();
            break;
        } else if (!status->vaild()) {
            hasnvaild = true;
            continue;
        }

        if (status->pid() == sta.pid()) {
            haschange = true;
            boost::unique_lock<boost::mutex> lock(m_mutex);
            *status = sta;
            lock.unlock();
            break;
        }
    }

    if (!haschange) {
        if (hasnvaild) {
            for (int i = 0; i < m_infoStruct.status_size(); i++) {
                CoreStatus *status = m_infoStruct.mutable_status(i);
                if (!status->vaild()) {
                    boost::unique_lock<boost::mutex> lock(m_mutex);
                    *status = sta;
                    lock.unlock();
                    break;
                }
            }
        } else {
            boost::unique_lock<boost::mutex> lock(m_mutex);
            CoreStatus *status = m_infoStruct.add_status();
            *status = sta;
            lock.unlock();
        }
    }

    int vaildcs = 0;
    int sTotalHasWorked = 0;
    int dTotalHasWorked = 0;
    int sCurrentPageCount = 0;
    int dCurrentPageCount = 0;
    int staticcs = 0;
    int dynamiccs = 0;
    for (int i = 0; i < m_infoStruct.status_size(); i++) {
        CoreStatus status = m_infoStruct.status(i);
        if (!status.vaild())
            continue;
        if (status.isstatic()) {
            staticcs ++;
            sTotalHasWorked += status.workedcount();
            sCurrentPageCount += status.runningcount() + status.waitcount();
        } else {
            dynamiccs ++;
            dTotalHasWorked += status.workedcount();
            dCurrentPageCount += status.runningcount() + status.waitcount();
        }
        vaildcs ++;
    }

    boost::unique_lock<boost::mutex> lock(m_mutex);
    m_infoStruct.set_scoresize(staticcs);
    m_infoStruct.set_dcoresize(dynamiccs);
    m_infoStruct.set_vaildcs(vaildcs);
    m_infoStruct.set_scurrentpc(sCurrentPageCount);
    m_infoStruct.set_dcurrentpc(dCurrentPageCount);
    m_infoStruct.set_soverflowpc(sTotalHasWorked - (configuer()->exitworkcount() * configuer()->staticcorecount()));
    m_infoStruct.set_doverflowpc(dTotalHasWorked - (configuer()->exitworkcount() * configuer()->dynamiccorecount()));
    lock.unlock();

    updateMemInfo();
}

void ManagerInfo::removeCoreStatus(int cpid)
{
    boost::unique_lock<boost::mutex> lock(m_mutex);
    for (int i = 0; i < m_infoStruct.status_size(); i++) {
        CoreStatus *status = m_infoStruct.mutable_status(i);
        if (status->pid() == cpid) {
            if (status->vaild()) {
                status->set_vaild(false);
            }
            m_needDestroyProcess.push_back(status->pid());
            break;
        }
    }
}

void ManagerInfo::setCoreSize(int size)
{
    boost::unique_lock<boost::mutex> lock(m_mutex);
    m_infoStruct.set_coresize(size);
}

void ManagerInfo::setConnectSize(int size)
{
    boost::unique_lock<boost::mutex> lock(m_mutex);
    m_infoStruct.set_connectsize(size);
}

int ManagerInfo::getFreePid(bool isStatic)
{
    double lowlevel = -1.0;
    int pid = -1;
    CoreStatus *lowstatus = NULL;

    for (int i = 0; i < m_infoStruct.status_size(); i++) {
        CoreStatus *status = m_infoStruct.mutable_status(i);
        if (!status->vaild())
            continue;
        if (status->state() != CoreStatus::Activited)
            continue;
        if (status->isstatic() != isStatic)
            continue;
        double level = status->level();
        if (lowlevel == -1.0) {
            lowlevel = level;
            pid = status->pid();
            lowstatus = status;
        } else if (level < lowlevel) {
            lowlevel = level;
            pid = status->pid();
            lowstatus = status;
        }
    }
    if (lowstatus) {
        boost::unique_lock<boost::mutex> lock(m_mutex);
        lowstatus->set_level(lowstatus->level() + 1.0);
        lock.unlock();
    }

    return pid;
}

int ManagerInfo::parseConfiguer(std::string content)
{
    Json::Reader reader;
    Json::Value value;
    if (!reader.parse(content.c_str(), value)) {
        fprintf(stderr, "Configuer file parse error!\n");
        return -1;
    }

    if (!value[GROUPNAME].isNull())
        configuer()->set_groupname(value[GROUPNAME].asString());
    else
        return -1;

    if (!value[STATICCORECOUNT].isNull())
        configuer()->set_staticcorecount(value[STATICCORECOUNT].asInt());
    else
        return -1;

    if (!value[DYNAMICCORECOUNT].isNull())
        configuer()->set_dynamiccorecount(value[DYNAMICCORECOUNT].asInt());
    else
        return -1;

    if (!value[ISDEBUG].isNull())
        configuer()->set_isdebug(value[ISDEBUG].asBool());
    else
        return -1;

    if (!value[ZOOKEEPERHOST].isNull())
        configuer()->set_zookeeperhost(value[ZOOKEEPERHOST].asString());
    else
        return -1;

    if (!value[EXITWORKCOUNT].isNull())
        configuer()->set_exitworkcount(value[EXITWORKCOUNT].asInt());
    else
        configuer()->set_exitworkcount(0);

    if (!value[DEFAULTSOCKETPORT].isNull())
        configuer()->set_socketport(value[DEFAULTSOCKETPORT].asString());
    else
        configuer()->set_socketport(DEFAULTPORT);

    if (!value[ENABLEPROXY].isNull() && value[ENABLEPROXY].asBool() == true) {
        m_enableProxy = true;
        m_proxyType = value[PROXYTYPE].asString();
        m_proxyAuthUser = value[PROXYAUTHUSER].asString();
        m_proxyAuthPass = value[PROXYAUTHPASS].asString();
        m_proxyHost = value[PROXYHOST].asString();
        m_proxyPort = value[PROXYPORT].asInt();
    }

    if (!value[ENABLEPROXYPOOL].isNull() && value[ENABLEPROXYPOOL].asBool() == true) {
        m_proxyPoolPath = value[PROXYPOOLPATH].asString();
        m_enableProxyPool = true;
        connectProxyPool();
    }

    return 0;
}

void ManagerInfo::updateMemInfo()
{
    struct sysinfo sysInfo;
    if (!sysinfo(&sysInfo)) {
        double percent = ((double)sysInfo.freeram/(double)sysInfo.totalram)*100;
        boost::unique_lock<boost::mutex> lock(m_mutex);
        m_infoStruct.set_freemem(sysInfo.freeram);
        m_infoStruct.set_usedmem(sysInfo.totalram - sysInfo.freeram);
        m_infoStruct.set_mempercent(percent);
        lock.unlock();
    }
}

std::string ManagerInfo::getIPAddress()
{
    int sock_fd;
    struct ifconf conf;
    struct ifreq *ifr;
    struct sockaddr_in *sin;
    char buff[128] = {0};
    int num, i;
    std::string ipAddress;

    sock_fd = socket(PF_INET, SOCK_DGRAM, 0);
    if(sock_fd < 0){
        perror("socket");
        return "";
    }
    conf.ifc_len = 128;
    conf.ifc_buf = buff;

    ioctl(sock_fd, SIOCGIFCONF, &conf);
    num = conf.ifc_len / sizeof(struct ifreq);
    ifr = conf.ifc_req;

    for(i = 0; i < num; i++){
        sin = (struct sockaddr_in *)(&ifr->ifr_addr);

        ioctl(sock_fd, SIOCGIFFLAGS, ifr);
        if(((ifr->ifr_flags & IFF_LOOPBACK) == 0) && (ifr->ifr_flags & IFF_UP)){
            ipAddress = inet_ntoa(sin->sin_addr);
            break;
        }
        ifr++;
    }

    close(sock_fd);

    return ipAddress;
}

void ManagerInfo::connectProxyPool()
{
    std::string host = configuer()->zookeeperhost();
    m_zkhandle = zookeeper_init(host.c_str(), zk_proxy_pool_watcher_g, 30000, 0, (void*)"spider proxy pool", 0);

    if (m_zkhandle == NULL) {
        fprintf(stderr, "Error when connecting to zookeeper servers...\n");
        exit(EXIT_FAILURE);
    }

    zoo_aget(m_zkhandle, m_proxyPoolPath.c_str(), 1, zk_proxy_pool_data_watcher_g, NULL);
}

std::string ManagerInfo::serialize()
{
    updateMemInfo();
    std::string statemess;
    boost::unique_lock<boost::mutex> lock(m_mutex);
    m_infoStruct.SerializeToString(&statemess);
    lock.unlock();
    return statemess;
}

std::string ManagerInfo::serializePageInfo()
{
    std::string statemess;
    PageInfoStruct info;

    boost::unique_lock<boost::mutex> lock(m_mutex);
    std::map<int, CorePages>::iterator it = m_pageInfoMap.begin();
    for (; it != m_pageInfoMap.end(); it ++) {
        CorePages *pages = info.add_pages();
        *pages = it->second;
    }
    lock.unlock();

    info.SerializeToString(&statemess);
    return statemess;
}

bool ManagerInfo::inNeedDestroyProcess(int cpid)
{
    for(std::list<int>::iterator it = m_needDestroyProcess.begin(); it != m_needDestroyProcess.end(); it++)
        if ((*it) == cpid)
            return true;
    return false;
}

void ManagerInfo::removeFromNeedDestroyProcessList(int cpid)
{
    m_needDestroyProcess.remove(cpid);
}

std::string ManagerInfo::getProxyFromPool()
{
    int tempPos = m_proxyPoolPos;

    m_proxyPoolPos ++;

    std::list<std::string>::iterator it = m_proxyPoolList.begin();
    for ( ; it != m_proxyPoolList.end(); it ++) {

        if (!(*it).empty() && (*it)!= "" && (*it).find(":") != -1 && (*it).find(".") != -1) {
            tempPos --;
        }

        if (tempPos < 0)
            return (*it);
    }

    m_proxyPoolPos = 0;

    return "";
}

void ManagerInfo::setProxyPool(std::list<std::string> proxys)
{
    m_proxyPoolList.clear();
    m_proxyPoolList = proxys;
}

}

